package com.study.spark.scala.structured_streaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.{ProcessingTime, Trigger}

/**
  *
  * @author stephen
  * @create 2019-03-03 11:14
  * @since 1.0.0
  */
object StructuredNetworkToMySQLWordCount {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[3]")
      .appName("StructuredNetworkToMySQLWordCount")
      .getOrCreate()

    import spark.implicits._

    // Create DataFrame representing the stream of input lines from connection to localhost:9999
    val lines = spark.readStream
      .format("socket")
      .option("host", "localhost")
      .option("port", 9999)
      .load()

    val words = lines
      .selectExpr("CAST(value AS STRING)")
      .as[String]
      .flatMap(_.split(" "))

    val result = words
      .groupBy("value")
      .count()
      .toDF("word", "count")

    /**
      *
      */
    val url = "jdbc:mysql://localhost:3306/test"
    val username = "root"
    val password = "123456"
    val sink = new JDBCSink(url, username, password)

    val query = result.writeStream
      .foreach(sink)
      .outputMode("update")
      .trigger(Trigger.ProcessingTime(5000))
      .start()

    query.awaitTermination()

  }

  //case class WordCount(word:String,count:Int)
}
